package com.fanxuankai.boot.canal.client;

import com.fanxuankai.boot.canal.client.autoconfigure.CanalClientProperties;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import javax.annotation.Resource;
import java.util.HashSet;
import java.util.Set;

/**
 * @author fanxuankai
 */
public class RabbitMqEntryConsumer implements EntryConsumer<MessageInfo> {
    @Resource
    private RabbitTemplate rabbitTemplate;
    @Resource
    private AmqpAdmin amqpAdmin;
    @Resource
    private Exchange exchange;
    @Resource
    private CanalClientProperties properties;
    private final Set<String> queueCache = new HashSet<>();

    @Override
    public void accept(MessageInfo messageInfo) {
        if (!queueCache.contains(messageInfo.getTopic())) {
            Queue queue = new Queue(messageInfo.getTopic());
            amqpAdmin.declareQueue(queue);
            amqpAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(messageInfo.getTopic()).noargs());
            queueCache.add(messageInfo.getTopic());
        }
        messageInfo.getMessages().forEach(s ->
                rabbitTemplate.convertAndSend(exchange.getName(), messageInfo.getTopic(), s));
    }

    @Override
    public MessageInfo apply(EntryWrapper entryWrapper) {
        return new MessageInfo(entryWrapper, properties);
    }
}
